package com.youzhi.message.kafka;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Map;

@Slf4j
public class KafkaMessageConsumers implements Runnable {


    private String kafkaServer;

    public KafkaMessageConsumers(String kafkaServer) {
        this.kafkaServer = kafkaServer;
        new Thread(this::run).start();
    }

    public void run() {

        KafkaConfig kafkaConfig = new KafkaConfig(this.kafkaServer);

        KafkaConsumer<String, String> kafkaConsumer = kafkaConfig.getKafkaConsumer();

        while (true) {

            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));

            if (consumerRecords != null && !consumerRecords.isEmpty()) {
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    log.info("收到kafka消息 key={},value={}", consumerRecord.key(), consumerRecord.value());
                    IKafkaMessageHandler handler = KafkaMessageHandlerContext.select(consumerRecord.key());
                    JSONObject jsonObject = JSONUtil.parseObj(consumerRecord.value());
                    Map<String, Object> map = BeanUtil.beanToMap(jsonObject.get("data"));
                    handler.handle(map);
                }
            }

        }


    }

}
